package org.idea.qiyu.framework.canal.handle;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.InvalidProtocolBufferException;
import org.idea.qiyu.framework.canal.config.CanalConfiguration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;

/**
 * @Author idea
 * @Date created in 10:11 上午 2020/8/2
 */
public class AbstractCanalHandler implements ICanalHandler {

    private static final Logger LOGGER = LoggerFactory.getLogger(AbstractCanalHandler.class);

    private String instanceName;

    private String tableName;

    public String getInstanceName() {
        return instanceName;
    }

    public void setInstanceName(String instanceName) {
        this.instanceName = instanceName;
    }

    public String getTableName() {
        return tableName;
    }

    public void setTableName(String tableName) {
        this.tableName = tableName;
    }


    private void dataHandle(List<CanalEntry.Entry> entryList) throws InvalidProtocolBufferException {
        for (CanalEntry.Entry entry : entryList) {
            if (entry.getEntryType() == CanalEntry.EntryType.ROWDATA) {
                dataEntryHandle(entry);
            }
        }
    }

    private void dataEntryHandle(CanalEntry.Entry entry) throws InvalidProtocolBufferException {
        String currentBinlogTableName = entry.getHeader().getTableName();
        if (!tableName.equals(currentBinlogTableName)) {
            return;
        }
        CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
        CanalEntry.EventType eventType = rowChange.getEventType();
        if (eventType == CanalEntry.EventType.DELETE) {
            this.onDelete(rowChange);
        } else if (eventType == CanalEntry.EventType.UPDATE) {
            this.onUpdate(rowChange);
        } else if (eventType == CanalEntry.EventType.INSERT) {
            this.onInsert(rowChange);
        }
    }

    @Override
    public void onDelete(CanalEntry.RowChange rowChange) {

    }

    @Override
    public void onUpdate(CanalEntry.RowChange rowChange) {

    }

    @Override
    public void onInsert(CanalEntry.RowChange rowChange) {

    }

    @Override
    public void consume(Message message) {
        try {
            this.dataHandle(message.getEntries());
        } catch (Exception e) {
            LOGGER.error("[AbstractCanalHandler] error is ", e);
        }
    }

}
